package com.atguigu.chapter07;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/3/9 9:33
 */
public class Flink26_State_Operator {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .socketTextStream("localhost", 9999)
                .map(new MyMapWithSumFunction())
                .print();

        env.execute();
    }

    public static class MyMapWithSumFunction implements MapFunction<String,Long>,CheckpointedFunction{

        ListState<Long> listState;
        private long sum = 0L;

        @Override
        public Long map(String value) throws Exception {
            long l = Long.parseLong(value);
            sum += l;
            listState.add(sum);
            return sum;
        }

        /**
         * 对状态做快照，可以在这里取出状态
         * @param context
         * @throws Exception
         */
        @Override
        public void snapshotState(FunctionSnapshotContext context) throws Exception {
            System.out.println("对状态做快照...");
        }

        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {
            listState = context.getOperatorStateStore().getListState(new ListStateDescriptor<Long>("listState", Types.LONG));
        }
    }
}
